package com.tc.zookeeper.curator;

import com.tc.zookeeper.MatchUtils;
import com.tc.zookeeper.ZKConstants;
import com.tc.zookeeper.ZKUtils;
import com.tc.zookeeper.exception.EnhancedCuratorException;
import lombok.Getter;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.ensemble.EnsembleProvider;
import org.apache.curator.ensemble.fixed.FixedEnsembleProvider;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.CuratorTempFramework;
import org.apache.curator.framework.api.ACLProvider;
import org.apache.curator.framework.api.CompressionProvider;
import org.apache.curator.framework.api.PathAndBytesable;
import org.apache.curator.framework.imps.DefaultACLProvider;
import org.apache.curator.framework.imps.GzipCompressionProvider;
import org.apache.curator.retry.RetryNTimes;
import org.apache.curator.utils.DefaultZookeeperFactory;
import org.apache.curator.utils.ZookeeperFactory;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

/**
 * 
 * EnhancedCuratorFramework Factory
 * 
 * @author owen.lu
 *
 */
public class EnhancedCuratorFrameworkFactory implements ZKConstants {

	private static final Logger LOGGER = LoggerFactory.getLogger(EnhancedCuratorFrameworkFactory.class);

	private static String bootstrapData;

	@Getter
	private static Map<String, EnhancedCuratorFramework> cachedCuratorFrameworkMap = new ConcurrentHashMap<String, EnhancedCuratorFramework>();
	
	/** key为partition，value为EnhancedCuratorFramework实例*/
	@Getter
	private static Map<String, Map<String, EnhancedCuratorFramework>> partitionEnhancedCuratorMaps = new ConcurrentHashMap<String, Map<String, EnhancedCuratorFramework>>();

	private static final byte[] LOCAL_ADDRESS = getLocalAddress();

	private static final CompressionProvider DEFAULT_COMPRESSION_PROVIDER = new GzipCompressionProvider();
	private static final DefaultZookeeperFactory DEFAULT_ZOOKEEPER_FACTORY = new DefaultZookeeperFactory();
	private static final DefaultACLProvider DEFAULT_ACL_PROVIDER = new DefaultACLProvider();
	private static final long DEFAULT_INACTIVE_THRESHOLD_MS = (int) TimeUnit.MINUTES.toMillis(3);

	public static EnhancedCuratorFramework newClient(String path) throws EnhancedCuratorException {
		String connectString = bootstrapTargetConnectString(path,-1,-1,null);//bootstrap中的data
		return newClientByConnectString(connectString);
	}

	public static EnhancedCuratorFramework newClient(String path, int sessionTimeoutMs, int connectionTimeoutMs)
			throws EnhancedCuratorException {
		String connectString = bootstrapTargetConnectString(path,sessionTimeoutMs,connectionTimeoutMs,null);//bootstrap中的data
		return newClientByConnectString(connectString, sessionTimeoutMs, connectionTimeoutMs);
	}

	public static EnhancedCuratorFramework newClient(String path, int sessionTimeoutMs, int connectionTimeoutMs,
			RetryPolicy retryPolicy) throws EnhancedCuratorException {
		String connectString = bootstrapTargetConnectString(path,sessionTimeoutMs,connectionTimeoutMs,retryPolicy);//bootstrap中的data
		return newClientByConnectString(connectString, sessionTimeoutMs, connectionTimeoutMs, retryPolicy);
	}

	public static EnhancedCuratorFramework newClientByConnectString(String connectString)
			throws EnhancedCuratorException {
		int sessionTimeoutMs = NumberUtils.toInt(ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_SESSION_IMEOUT_MS),
				DEFAULT_SESSION_TIMEOUT_MS);
		int connectionTimeoutMs = NumberUtils.toInt(
				ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_CONNECTION_IMEOUT_MS), DEFAULT_CONNECTION_TIMEOUT_MS);
		return newClientByConnectString(connectString, sessionTimeoutMs, connectionTimeoutMs);
	}

	public static EnhancedCuratorFramework newClientByConnectString(String connectString, int sessionTimeoutMs,
			int connectionTimeoutMs) throws EnhancedCuratorException {
		int nTimes = NumberUtils.toInt(ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_MAXRETRIES), DEFAULT_RETRY_TIMES);
		int sleepMsBetweenRetries = NumberUtils.toInt(
				ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_SLEEPMSBETWEENRETRY), DEFAULT_SLEEPMS_BETWEENRETRY);
		RetryPolicy retryPolicy = new RetryNTimes(nTimes, sleepMsBetweenRetries);
		return newClientByConnectString(connectString, sessionTimeoutMs, connectionTimeoutMs, retryPolicy);
	}

	public static EnhancedCuratorFramework newClientByConnectString(String connectString, int sessionTimeoutMs,
			int connectionTimeoutMs, RetryPolicy retryPolicy) throws EnhancedCuratorException {
		return newClientByPartitionConnectString(getEnvPartition(), connectString, sessionTimeoutMs,
				connectionTimeoutMs, retryPolicy);
	}

	public static EnhancedCuratorFramework newClientByPartitionConnectString(String partition, String connectString,
			int sessionTimeoutMs, int connectionTimeoutMs, RetryPolicy retryPolicy) throws EnhancedCuratorException {
		//根据同机房优先权重策略来拼装ip
		String connectStringByWeight = MatchUtils.getConnectStringByWeightStragety(connectString.trim());
		return initEnhancedCuratorFramework(partition, connectStringByWeight, sessionTimeoutMs, connectionTimeoutMs,
				retryPolicy);
	}

	public static EnhancedCuratorFramework newClientByDomain(String domain) throws EnhancedCuratorException {
		return newClientByDomain(getEnvPartition(), domain);
	}

	public static EnhancedCuratorFramework newClientByDomain(String domain, int sessionTimeoutMs,
			int connectionTimeoutMs) throws EnhancedCuratorException {
		return newClientByDomain(getEnvPartition(), domain, sessionTimeoutMs, connectionTimeoutMs);
	}

	public static EnhancedCuratorFramework newClientByDomain(String domain, int sessionTimeoutMs,
			int connectionTimeoutMs, RetryPolicy retryPolicy) throws EnhancedCuratorException {
		return newClientByDomain(getEnvPartition(), domain, sessionTimeoutMs, connectionTimeoutMs, retryPolicy);
	}

	public static EnhancedCuratorFramework newClientByDomain(String partition, String domain)
			throws EnhancedCuratorException {
		int sessionTimeoutMs = NumberUtils.toInt(ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_SESSION_IMEOUT_MS),
				DEFAULT_SESSION_TIMEOUT_MS);
		int connectionTimeoutMs = NumberUtils.toInt(
				ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_CONNECTION_IMEOUT_MS), DEFAULT_CONNECTION_TIMEOUT_MS);
		return newClientByDomain(partition, domain, sessionTimeoutMs, connectionTimeoutMs);
	}

	public static EnhancedCuratorFramework newClientByDomain(String partition, String domain, int sessionTimeoutMs,
			int connectionTimeoutMs) throws EnhancedCuratorException {
		int nTimes = NumberUtils.toInt(ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_MAXRETRIES), DEFAULT_RETRY_TIMES);
		int sleepMsBetweenRetries = NumberUtils.toInt(
				ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_SLEEPMSBETWEENRETRY), DEFAULT_SLEEPMS_BETWEENRETRY);
		RetryPolicy retryPolicy = new RetryNTimes(nTimes, sleepMsBetweenRetries);
		return newClientByDomain(partition, domain, sessionTimeoutMs, connectionTimeoutMs, retryPolicy);
	}

	public static EnhancedCuratorFramework newClientByDomain(String partition, String domain, int sessionTimeoutMs,
			int connectionTimeoutMs, RetryPolicy retryPolicy) throws EnhancedCuratorException {
		//domain中的data
		String connectStringByDomain = getTargetConnectStringByDomain(partition, domain, sessionTimeoutMs, connectionTimeoutMs, retryPolicy);
		//根据同机房优先权重策略拼装ip
		String connectStringByWeight = MatchUtils.getConnectStringByWeightStragety(connectStringByDomain);
		return initEnhancedCuratorFramework(partition, connectStringByWeight, sessionTimeoutMs,
				connectionTimeoutMs, retryPolicy);
	}

	public static String getEnvPartition() {
		String envPartition = ZKUtils.resolveSystemProperty(VIP_CFGCENTER_PARTITION);
		if (StringUtils.isNotEmpty(envPartition)) {
			return envPartition;
		}
		return ZOOKEEPER_PARTITION_DEFAULT;
	}

	private static EnhancedCuratorFramework initEnhancedCuratorFramework(String partition, String connectString,
			int sessionTimeoutMs, int connectionTimeoutMs, RetryPolicy retryPolicy) throws EnhancedCuratorException {
		checkConnectString(connectString);
		EnhancedCuratorFramework enhancedCuratorFramework = builder().connectString(connectString).sessionTimeoutMs(sessionTimeoutMs)
				.connectionTimeoutMs(connectionTimeoutMs).retryPolicy(retryPolicy).namespace(partition).build();
		((EnhancedCuratorFrameworkImpl) enhancedCuratorFramework).setTargetConnectString(connectString);
		return enhancedCuratorFramework;
	}

	private static EnhancedCuratorFramework initTempEnhancedCuratorFramework(String connectString, String partition,int sessionTimeoutMs,int connectionTimeoutMs, RetryPolicy retryPolicy)
			throws EnhancedCuratorException {
		if(-1 == sessionTimeoutMs){
			sessionTimeoutMs = NumberUtils.toInt(ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_SESSION_IMEOUT_MS),
					DEFAULT_SESSION_TIMEOUT_MS);
		}
        if(-1 == connectionTimeoutMs){
        	connectionTimeoutMs = NumberUtils.toInt(
					ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_CONNECTION_IMEOUT_MS), DEFAULT_CONNECTION_TIMEOUT_MS);
		}
		if(null == retryPolicy){
			int nTimes = NumberUtils.toInt(ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_MAXRETRIES), DEFAULT_RETRY_TIMES);
			int sleepMsBetweenRetries = NumberUtils.toInt(
					ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_SLEEPMSBETWEENRETRY), DEFAULT_SLEEPMS_BETWEENRETRY);
			retryPolicy = new RetryNTimes(nTimes, sleepMsBetweenRetries);
		}
		return initEnhancedCuratorFramework(partition, connectString, sessionTimeoutMs,
				connectionTimeoutMs, retryPolicy);
	}

	public static EnhancedCuratorFramework getStartedClientByConnectString(String connectString)
			throws EnhancedCuratorException {
		int sessionTimeoutMs = NumberUtils.toInt(ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_SESSION_IMEOUT_MS),
				DEFAULT_SESSION_TIMEOUT_MS);
		int connectionTimeoutMs = NumberUtils.toInt(
				ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_CONNECTION_IMEOUT_MS), DEFAULT_CONNECTION_TIMEOUT_MS);
		return getStartedClientByConnectString(connectString, sessionTimeoutMs, connectionTimeoutMs);
	}

	public static EnhancedCuratorFramework getStartedClientByConnectString(String connectString, int sessionTimeoutMs,
			int connectionTimeoutMs) throws EnhancedCuratorException {
		int nTimes = NumberUtils.toInt(ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_MAXRETRIES), DEFAULT_RETRY_TIMES);
		int sleepMsBetweenRetries = NumberUtils.toInt(
				ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_SLEEPMSBETWEENRETRY), DEFAULT_SLEEPMS_BETWEENRETRY);
		RetryPolicy retryPolicy = new RetryNTimes(nTimes, sleepMsBetweenRetries);
		return getStartedClientByConnectString(connectString, sessionTimeoutMs, connectionTimeoutMs, retryPolicy);
	}

	public static EnhancedCuratorFramework getStartedClientByConnectString(String connectString, int sessionTimeoutMs,
			int connectionTimeoutMs, RetryPolicy retryPolicy) throws EnhancedCuratorException {
		return getSynchronizedClientByPartitionConnectString(getEnvPartition(), connectString, sessionTimeoutMs,
				connectionTimeoutMs, retryPolicy);
	}

	public static EnhancedCuratorFramework getStartedClient(String path) throws EnhancedCuratorException {
		String targetConnectString = bootstrapTargetConnectString(path,-1,-1,null);
		return getStartedClientByConnectString(targetConnectString);
	}

	public static EnhancedCuratorFramework getStartedClient(String path, int sessionTimeoutMs, int connectionTimeoutMs)
			throws EnhancedCuratorException {
		String targetConnectString = bootstrapTargetConnectString(path,sessionTimeoutMs,connectionTimeoutMs,null);
		return getStartedClientByConnectString(targetConnectString, sessionTimeoutMs, connectionTimeoutMs);
	}

	public static EnhancedCuratorFramework getStartedClient(String path, int sessionTimeoutMs, int connectionTimeoutMs,
			RetryPolicy retryPolicy) throws EnhancedCuratorException {
		String targetConnectString = bootstrapTargetConnectString(path,sessionTimeoutMs,connectionTimeoutMs,retryPolicy);
		return getStartedClientByConnectString(targetConnectString, sessionTimeoutMs, connectionTimeoutMs, retryPolicy);
	}
	
	public static EnhancedCuratorFramework getStartedClientByPartition(String path, String partition) throws EnhancedCuratorException {
		String targetConnectString = bootstrapTargetConnectStringByPartition(path,-1,-1,null,partition);
		return getStartedClientByConnectStringByPartition(targetConnectString,partition);
	}

	public static EnhancedCuratorFramework getStartedClientByPartition(String path, String partition, int sessionTimeoutMs, int connectionTimeoutMs)
			throws EnhancedCuratorException {
		String targetConnectString = bootstrapTargetConnectStringByPartition(path,sessionTimeoutMs,connectionTimeoutMs,null,partition);
		return getStartedClientByConnectStringByPartition(targetConnectString, sessionTimeoutMs, connectionTimeoutMs, partition);
	}

	public static EnhancedCuratorFramework getStartedClientByPartition(String path, String partition, int sessionTimeoutMs, int connectionTimeoutMs,
			RetryPolicy retryPolicy) throws EnhancedCuratorException {
		String targetConnectString = bootstrapTargetConnectStringByPartition(path,sessionTimeoutMs,connectionTimeoutMs,retryPolicy,partition);
		return getStartedClientByConnectStringByPartition(targetConnectString, sessionTimeoutMs, connectionTimeoutMs, retryPolicy, partition);
	}

	public static EnhancedCuratorFramework getStartedClientByDomain(String domain) throws EnhancedCuratorException {
		return getStartedClientByDomain(getEnvPartition(), domain);
	}
	
	public static EnhancedCuratorFramework getStartedClientByConnectStringByPartition(String connectString, String partition)
			throws EnhancedCuratorException {
		int sessionTimeoutMs = NumberUtils.toInt(ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_SESSION_IMEOUT_MS),
				DEFAULT_SESSION_TIMEOUT_MS);
		int connectionTimeoutMs = NumberUtils.toInt(
				ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_CONNECTION_IMEOUT_MS), DEFAULT_CONNECTION_TIMEOUT_MS);
		return getStartedClientByConnectStringByPartition(connectString, sessionTimeoutMs, connectionTimeoutMs,partition);
	}

	public static EnhancedCuratorFramework getStartedClientByConnectStringByPartition(String connectString, int sessionTimeoutMs,
			int connectionTimeoutMs, String partition) throws EnhancedCuratorException {
		int nTimes = NumberUtils.toInt(ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_MAXRETRIES), DEFAULT_RETRY_TIMES);
		int sleepMsBetweenRetries = NumberUtils.toInt(
				ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_SLEEPMSBETWEENRETRY), DEFAULT_SLEEPMS_BETWEENRETRY);
		RetryPolicy retryPolicy = new RetryNTimes(nTimes, sleepMsBetweenRetries);
		return getStartedClientByConnectStringByPartition(connectString, sessionTimeoutMs, connectionTimeoutMs, retryPolicy, partition);
	}

	public static EnhancedCuratorFramework getStartedClientByConnectStringByPartition(String connectString, int sessionTimeoutMs,
                                                                                      int connectionTimeoutMs, RetryPolicy retryPolicy, String partition) throws EnhancedCuratorException {
		return getSynchronizedClientByPartition(partition, connectString, sessionTimeoutMs,
				connectionTimeoutMs, retryPolicy);
	}

	public static EnhancedCuratorFramework getStartedClientByDomain(String domain, int sessionTimeoutMs,
			int connectionTimeoutMs) throws EnhancedCuratorException {
		return getStartedClientByDomain(getEnvPartition(), domain, sessionTimeoutMs, connectionTimeoutMs);
	}

	public static EnhancedCuratorFramework getStartedClientByDomain(String domain, int sessionTimeoutMs,
			int connectionTimeoutMs, RetryPolicy retryPolicy) throws EnhancedCuratorException {
		return getStartedClientByDomain(getEnvPartition(), domain, sessionTimeoutMs, connectionTimeoutMs, retryPolicy);
	}

	public static EnhancedCuratorFramework getStartedClientByDomain(String partition, String domain)
			throws EnhancedCuratorException {
		int sessionTimeoutMs = NumberUtils.toInt(ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_SESSION_IMEOUT_MS),
				DEFAULT_SESSION_TIMEOUT_MS);
		int connectionTimeoutMs = NumberUtils.toInt(
				ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_CONNECTION_IMEOUT_MS), DEFAULT_CONNECTION_TIMEOUT_MS);
		return getStartedClientByDomain(partition, domain, sessionTimeoutMs, connectionTimeoutMs);
	}

	public static EnhancedCuratorFramework getStartedClientByDomain(String partition, String domain,
			int sessionTimeoutMs, int connectionTimeoutMs) throws EnhancedCuratorException {
		int nTimes = NumberUtils.toInt(ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_MAXRETRIES), DEFAULT_RETRY_TIMES);
		int sleepMsBetweenRetries = NumberUtils.toInt(
				ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_SLEEPMSBETWEENRETRY), DEFAULT_SLEEPMS_BETWEENRETRY);
		RetryPolicy retryPolicy = new RetryNTimes(nTimes, sleepMsBetweenRetries);
		return getStartedClientByDomain(partition, domain, sessionTimeoutMs, connectionTimeoutMs, retryPolicy);
	}

	public static EnhancedCuratorFramework getStartedClientByDomain(String partition, String domain,
			int sessionTimeoutMs, int connectionTimeoutMs, RetryPolicy retryPolicy) throws EnhancedCuratorException {
		String connectStringByDomain = getTargetConnectStringByDomain(partition, domain, sessionTimeoutMs, connectionTimeoutMs, retryPolicy);//domain中的data
		return getSynchronizedClientByPartitionConnectString(partition, connectStringByDomain, sessionTimeoutMs,
				connectionTimeoutMs, retryPolicy);
	}

	public static EnhancedCuratorFramework getSynchronizedClientByPartitionConnectString(final String partition,
			final String connectString, final int sessionTimeoutMs, final int connectionTimeoutMs,
			final RetryPolicy retryPolicy) throws EnhancedCuratorException {
		checkConnectString(connectString);

		EnhancedCuratorFramework enhancedCuratorFramework = cachedCuratorFrameworkMap.get(connectString);
		if (null == enhancedCuratorFramework) {
			synchronized (EnhancedCuratorFrameworkFactory.class) {
				enhancedCuratorFramework = newClientByPartitionConnectString(partition, connectString,
						sessionTimeoutMs, connectionTimeoutMs, retryPolicy);
				//((EnhancedCuratorFrameworkImpl) enhancedCuratorFramework).setTargetConnectString(connectString);

				enhancedCuratorFramework.start();

				cachedCuratorFrameworkMap.put(connectString, enhancedCuratorFramework);
			}
		}

		return enhancedCuratorFramework;
	}
	
	//add for partition connect
	public static EnhancedCuratorFramework getSynchronizedClientByPartition(final String partition,
			final String connectString, final int sessionTimeoutMs, final int connectionTimeoutMs,
			final RetryPolicy retryPolicy) throws EnhancedCuratorException {
		checkConnectString(connectString);
		
		Map<String,EnhancedCuratorFramework> enhancedCuratorFrameworkMap = partitionEnhancedCuratorMaps.get(partition);
		
		EnhancedCuratorFramework enhancedCuratorFramework = null;
		if(enhancedCuratorFrameworkMap != null) {
			enhancedCuratorFramework = enhancedCuratorFrameworkMap.get(connectString);
		}
		
		if (null == enhancedCuratorFramework) {
			synchronized (EnhancedCuratorFrameworkFactory.class) {
				enhancedCuratorFramework = newClientByPartitionConnectString(partition, connectString,
						sessionTimeoutMs, connectionTimeoutMs, retryPolicy);
				//((EnhancedCuratorFrameworkImpl) enhancedCuratorFramework).setTargetConnectString(connectString);

				enhancedCuratorFramework.start();

				cachedCuratorFrameworkMap.put(connectString, enhancedCuratorFramework);
				partitionEnhancedCuratorMaps.put(partition, cachedCuratorFrameworkMap);
			}
		}

		return enhancedCuratorFramework;
	}

	public static String getTargetConnectStringByDomain(String partition, String domain,int sessionTimeoutMs,
			int connectionTimeoutMs, RetryPolicy retryPolicy)
			throws EnhancedCuratorException {
		String connectStringBydomain = domainTargetConnectString(partition, domain,sessionTimeoutMs,connectionTimeoutMs,retryPolicy); //domain中的data
		return connectStringBydomain;
	}
	

	public static String bootstrapTargetConnectString(String path,int sessionTimeoutMs,
			int connectionTimeoutMs, RetryPolicy retryPolicy) throws EnhancedCuratorException {
		Map<String, Object> zkPropsMap = ZKUtils.propertiesToMap(getBootstrapData(sessionTimeoutMs,connectionTimeoutMs,retryPolicy));
		for (Map.Entry<String, Object> entry : zkPropsMap.entrySet()) {
			if (ZKUtils.isMatch(entry.getKey(), path)) {
				return entry.getValue().toString().trim();
			}
		}

		return ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_CONNECTION);
	}
	
	public static String bootstrapTargetConnectStringByPartition(String path, int sessionTimeoutMs,
                                                                 int connectionTimeoutMs, RetryPolicy retryPolicy, String partition) throws EnhancedCuratorException {
		Map<String, Object> zkPropsMap = ZKUtils.propertiesToMap(getBootstrapDataByPartition(sessionTimeoutMs,connectionTimeoutMs,retryPolicy,partition));
		for (Map.Entry<String, Object> entry : zkPropsMap.entrySet()) {
			if (ZKUtils.isMatch(entry.getKey(), path)) {
				return entry.getValue().toString().trim();
			}
		}

		return ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_CONNECTION);
	}
	
	public static String getBootstrapDataByPartition(int sessionTimeoutMs,
                                                     int connectionTimeoutMs, RetryPolicy retryPolicy, String partition) throws EnhancedCuratorException {
		String connectString = ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_CONNECTION);
		String targetString = MatchUtils.getConnectStringByWeightStragety(connectString);
		EnhancedCuratorFramework tempEnhancedCuratorFramework = initTempEnhancedCuratorFramework(
				targetString,partition,sessionTimeoutMs,connectionTimeoutMs, retryPolicy);
		try {
			tempEnhancedCuratorFramework.start();
			if (null != tempEnhancedCuratorFramework.checkExists().forPath("/" + BOOTSTRAP)) {
				byte[] bytes = tempEnhancedCuratorFramework.getData().forPath("/" + BOOTSTRAP);
				if (bytes != null) {
					bootstrapData = new String(bytes, DEFAULT_CHARSET);
					LOGGER.info("current bootstrapDataByPartition is >>>" + bootstrapData);
				}
			}
		} catch (Exception e) {
			throw new EnhancedCuratorException(e);
		} finally {
			if (tempEnhancedCuratorFramework != null) {
				tempEnhancedCuratorFramework.close();
			}
		}

		return bootstrapData;
	}

	public static String getBootstrapData( int sessionTimeoutMs,
			int connectionTimeoutMs, RetryPolicy retryPolicy) throws EnhancedCuratorException {
		String connectString = ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_CONNECTION);
		String targetString = MatchUtils.getConnectStringByWeightStragety(connectString);
		EnhancedCuratorFramework tempEnhancedCuratorFramework = initTempEnhancedCuratorFramework(
				targetString, getEnvPartition(),sessionTimeoutMs,connectionTimeoutMs, retryPolicy);
		try {
			tempEnhancedCuratorFramework.start();
			if (null != tempEnhancedCuratorFramework.checkExists().forPath("/" + BOOTSTRAP)) {
				byte[] bytes = tempEnhancedCuratorFramework.getData().forPath("/" + BOOTSTRAP);
				if (bytes != null) {
					bootstrapData = new String(bytes, DEFAULT_CHARSET);
					LOGGER.info("current bootstrapData is >>>" + bootstrapData);
				}
			}
		} catch (Exception e) {
			throw new EnhancedCuratorException(e);
		} finally {
			if (tempEnhancedCuratorFramework != null) {
				tempEnhancedCuratorFramework.close();
			}
		}

		return bootstrapData;
	}

	private static String domainTargetConnectString(String partition, String domain,int sessionTimeoutMs,
			int connectionTimeoutMs, RetryPolicy retryPolicy) throws EnhancedCuratorException {
		return getConnStringByDomainData(getDomainData(partition, domain, sessionTimeoutMs, connectionTimeoutMs, retryPolicy));
	}
   
	//get connect IP from domain data
	public static String getConnStringByDomainData(String domainData) {
		Map<String, Object> zkPropsMap = ZKUtils.jsonToMap(domainData);
		if (zkPropsMap != null && zkPropsMap.size() > 0) {
			return zkPropsMap.get(ZKConstants.ZKCONNECTION) + "";
		}
		return ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_CONNECTION);
	}

	public static String getDomainData(String connectString, String partition, String domain,int sessionTimeoutMs,
			int connectionTimeoutMs, RetryPolicy retryPolicy)
			throws EnhancedCuratorException {
		EnhancedCuratorFramework tempEnhancedCuratorFramework = initTempEnhancedCuratorFramework(connectString,
				partition,sessionTimeoutMs,connectionTimeoutMs,retryPolicy);
		try {
			tempEnhancedCuratorFramework.start();

			String clusterPath = null;
			final String domainZkPath = "/" + DOMAINS_MAPPING + "/" + domain;
			if (null != tempEnhancedCuratorFramework.checkExists().forPath(domainZkPath)) {
				byte[] bytes = tempEnhancedCuratorFramework.getData().forPath(domainZkPath);
				clusterPath = new String(bytes, DEFAULT_CHARSET);

				if (StringUtils.isBlank(clusterPath)) {
					throw new EnhancedCuratorException("zkdata in domainZkPath " + domainZkPath + " is null.");
				}
			}

			if (null == clusterPath) {
				final String defaultDomainPath = "/" + DOMAINS_MAPPING + "/" + DEFAULT_DOMAIN;
				if (null != tempEnhancedCuratorFramework.checkExists().forPath(defaultDomainPath)) {
					byte[] bytes = tempEnhancedCuratorFramework.getData().forPath(defaultDomainPath);
					clusterPath = new String(bytes, DEFAULT_CHARSET);

					if (StringUtils.isBlank(clusterPath)) {
						throw new EnhancedCuratorException("zkdata in defaultDomainPath " + defaultDomainPath
								+ " is null.");
					}
				}
			}

			if (StringUtils.isNotBlank(clusterPath)) {
				if (null != tempEnhancedCuratorFramework.checkExists().forPath(clusterPath)) {
					byte[] bytes = tempEnhancedCuratorFramework.getData().forPath(clusterPath);
					if (bytes != null && bytes.length > 0) {
						return new String(bytes, DEFAULT_CHARSET);
					} else {
						throw new EnhancedCuratorException("zkdata in clusterPath " + clusterPath + " is null.");
					}
				} else {
					throw new EnhancedCuratorException("clusterPath " + clusterPath + " does not exists.");
				}
			} else {
				throw new EnhancedCuratorException("clusterPath is null.");
			}
		} catch (Exception e) {
			throw new EnhancedCuratorException(e);
		} finally {
			if (tempEnhancedCuratorFramework != null) {
				tempEnhancedCuratorFramework.close();
			}
		}
	}
    //get data in domain
	public static String getDomainData(String partition, String domain,int sessionTimeoutMs,
			int connectionTimeoutMs, RetryPolicy retryPolicy) throws EnhancedCuratorException {
		String connectString = ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_CONNECTION);
		String targetString = MatchUtils.getConnectStringByWeightStragety(connectString);
		return getDomainData(targetString, partition, domain, sessionTimeoutMs, connectionTimeoutMs,retryPolicy);
	}

	private static void checkConnectString(String connectString) throws EnhancedCuratorException {
		if (StringUtils.isEmpty(connectString)) {
			throw new EnhancedCuratorException(
					"Initialize zookeeper failed, please config sys or env property VIP_CFGCENTER_ZK_CONNECTION");
		}
	}
	


	/**
	 * Return a new builder that builds a CuratorFramework
	 *
	 * @return new builder
	 */
	public static Builder builder() {
		return new Builder();
	}

	public static class Builder {
		private EnsembleProvider ensembleProvider;
		private int sessionTimeoutMs = ZKConstants.DEFAULT_SESSION_TIMEOUT_MS;
		private int connectionTimeoutMs = ZKConstants.DEFAULT_CONNECTION_TIMEOUT_MS;
		private RetryPolicy retryPolicy;
		private ThreadFactory threadFactory = null;
		private String namespace;
		private String authScheme = null;
		private byte[] authValue = null;
		private byte[] defaultData = LOCAL_ADDRESS;
		private CompressionProvider compressionProvider = DEFAULT_COMPRESSION_PROVIDER;
		private ZookeeperFactory zookeeperFactory = DEFAULT_ZOOKEEPER_FACTORY;
		private ACLProvider aclProvider = DEFAULT_ACL_PROVIDER;
		private boolean canBeReadOnly = false;

		/**
		 * Apply the current values and build a new CuratorFramework
		 *
		 * @return new CuratorFramework
		 */
		public EnhancedCuratorFramework build() {
			return new EnhancedCuratorFrameworkImpl(this);
		}

		/**
		 * Apply the current values and build a new temporary CuratorFramework. Temporary CuratorFramework instances are
		 * meant for single requests to ZooKeeper ensembles over a failure prone network such as a WAN. The APIs
		 * available from {@link CuratorTempFramework} are limited. Further, the connection will be closed after 3
		 * minutes of inactivity.
		 *
		 * @return temp instance
		 */
		public EnhancedCuratorTempFramework buildTemp() {
			return buildTemp(DEFAULT_INACTIVE_THRESHOLD_MS, TimeUnit.MILLISECONDS);
		}

		/**
		 * Apply the current values and build a new temporary CuratorFramework. Temporary CuratorFramework instances are
		 * meant for single requests to ZooKeeper ensembles over a failure prone network such as a WAN. The APIs
		 * available from {@link CuratorTempFramework} are limited. Further, the connection will be closed after
		 * <code>inactiveThresholdMs</code> milliseconds of inactivity.
		 *
		 * @param inactiveThreshold number of milliseconds of inactivity to cause connection close
		 * @param unit threshold unit
		 * @return temp instance
		 */
		public EnhancedCuratorTempFramework buildTemp(long inactiveThreshold, TimeUnit unit) {
			return new EnhancedCuratorTempFrameworkImpl(this, unit.toMillis(inactiveThreshold));
		}

		/**
		 * Add connection authorization
		 *
		 * @param scheme the scheme
		 * @param auth the auth bytes
		 * @return this
		 */
		public Builder authorization(String scheme, byte[] auth) {
			this.authScheme = scheme;
			this.authValue = (auth != null) ? Arrays.copyOf(auth, auth.length) : null;
			return this;
		}

		/**
		 * Set the list of servers to connect to. IMPORTANT: use either this or
		 * {@link #ensembleProvider(EnsembleProvider)} but not both.
		 *
		 * @param connectString list of servers to connect to
		 * @return this
		 */
		public Builder connectString(String connectString) {
			ensembleProvider = new FixedEnsembleProvider(connectString);
			return this;
		}

		/**
		 * Set the list ensemble provider. IMPORTANT: use either this or {@link #connectString(String)} but not both.
		 *
		 * @param ensembleProvider the ensemble provider to use
		 * @return this
		 */
		public Builder ensembleProvider(EnsembleProvider ensembleProvider) {
			this.ensembleProvider = ensembleProvider;
			return this;
		}

		/**
		 * Sets the data to use when {@link PathAndBytesable#forPath(String)} is used. This is useful for debugging
		 * purposes. For example, you could set this to be the IP of the client.
		 *
		 * @param defaultData new default data to use
		 * @return this
		 */
		public Builder defaultData(byte[] defaultData) {
			this.defaultData = (defaultData != null) ? Arrays.copyOf(defaultData, defaultData.length) : null;
			return this;
		}

		/**
		 * As ZooKeeper is a shared space, users of a given cluster should stay within a pre-defined namespace. If a
		 * namespace is set here, all paths will get pre-pended with the namespace
		 *
		 * @param namespace the namespace
		 * @return this
		 */
		public Builder namespace(String namespace) {
			this.namespace = namespace;
			return this;
		}

		/**
		 * @param sessionTimeoutMs session timeout
		 * @return this
		 */
		public Builder sessionTimeoutMs(int sessionTimeoutMs) {
			this.sessionTimeoutMs = sessionTimeoutMs;
			return this;
		}

		/**
		 * @param connectionTimeoutMs connection timeout
		 * @return this
		 */
		public Builder connectionTimeoutMs(int connectionTimeoutMs) {
			this.connectionTimeoutMs = connectionTimeoutMs;
			return this;
		}

		/**
		 * @param retryPolicy retry policy to use
		 * @return this
		 */
		public Builder retryPolicy(RetryPolicy retryPolicy) {
			this.retryPolicy = retryPolicy;
			return this;
		}

		/**
		 * @param threadFactory thread factory used to create Executor Services
		 * @return this
		 */
		public Builder threadFactory(ThreadFactory threadFactory) {
			this.threadFactory = threadFactory;
			return this;
		}

		/**
		 * @param compressionProvider the compression provider
		 * @return this
		 */
		public Builder compressionProvider(CompressionProvider compressionProvider) {
			this.compressionProvider = compressionProvider;
			return this;
		}

		/**
		 * @param zookeeperFactory the zookeeper factory to use
		 * @return this
		 */
		public Builder zookeeperFactory(ZookeeperFactory zookeeperFactory) {
			this.zookeeperFactory = zookeeperFactory;
			return this;
		}

		/**
		 * @param aclProvider a provider for ACLs
		 * @return this
		 */
		public Builder aclProvider(ACLProvider aclProvider) {
			this.aclProvider = aclProvider;
			return this;
		}

		/**
		 * @param canBeReadOnly if true, allow ZooKeeper client to enter read only mode in case of a network partition.
		 * See {@link ZooKeeper#ZooKeeper(String, int, Watcher, long, byte[], boolean)} for details
		 * @return this
		 */
		public Builder canBeReadOnly(boolean canBeReadOnly) {
			this.canBeReadOnly = canBeReadOnly;
			return this;
		}

		public ACLProvider getAclProvider() {
			return aclProvider;
		}

		public ZookeeperFactory getZookeeperFactory() {
			return zookeeperFactory;
		}

		public CompressionProvider getCompressionProvider() {
			return compressionProvider;
		}

		public ThreadFactory getThreadFactory() {
			return threadFactory;
		}

		public EnsembleProvider getEnsembleProvider() {
			return ensembleProvider;
		}

		public int getSessionTimeoutMs() {
			return sessionTimeoutMs;
		}

		public int getConnectionTimeoutMs() {
			return connectionTimeoutMs;
		}

		public RetryPolicy getRetryPolicy() {
			return retryPolicy;
		}

		public String getNamespace() {
			return namespace;
		}

		public String getAuthScheme() {
			return authScheme;
		}

		public byte[] getAuthValue() {
			return (authValue != null) ? Arrays.copyOf(authValue, authValue.length) : null;
		}

		public byte[] getDefaultData() {
			return defaultData;
		}

		public boolean canBeReadOnly() {
			return canBeReadOnly;
		}

		private Builder() {
		}

		public CuratorFrameworkFactory.Builder toCuratorBuilder() {
			CuratorFrameworkFactory.Builder curatorBuilder = CuratorFrameworkFactory.builder();
			curatorBuilder.aclProvider(this.aclProvider).canBeReadOnly(canBeReadOnly)
					.compressionProvider(compressionProvider).connectionTimeoutMs(connectionTimeoutMs)
					.defaultData(defaultData).ensembleProvider(ensembleProvider).namespace(namespace)
					.retryPolicy(retryPolicy).sessionTimeoutMs(sessionTimeoutMs).threadFactory(threadFactory)
					.zookeeperFactory(zookeeperFactory);
			return curatorBuilder;
		}

	}

	private static byte[] getLocalAddress() {
		try {
			return InetAddress.getLocalHost().getHostAddress().getBytes();
		} catch (UnknownHostException ignore) {
		}
		return new byte[0];
	}
	
	//old api
	private static volatile boolean haveFetchBootstrap = false;
	public static String getBootstrapData(Boolean refetchBootstrap) throws EnhancedCuratorException {
		if ((!haveFetchBootstrap && !refetchBootstrap)) {
			EnhancedCuratorFramework tempEnhancedCuratorFramework = initTempEnhancedCuratorFramework(
					ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_CONNECTION), getEnvPartition());
			try {
				tempEnhancedCuratorFramework.start();
				if (null != tempEnhancedCuratorFramework.checkExists().forPath("/" + BOOTSTRAP)) {
					byte[] bytes = tempEnhancedCuratorFramework.getData().forPath("/" + BOOTSTRAP);
					if (bytes != null) {
						bootstrapData = new String(bytes, DEFAULT_CHARSET);
						LOGGER.info("current bootstrapData is >>>" + bootstrapData);
					}
				}
				haveFetchBootstrap = true;
			} catch (Exception e) {
				throw new EnhancedCuratorException(e);
			} finally {
				if (tempEnhancedCuratorFramework != null) {
					tempEnhancedCuratorFramework.close();
				}
			}
		}

		return bootstrapData;
	}
	private static EnhancedCuratorFramework initTempEnhancedCuratorFramework(String connectString, String partition)
			throws EnhancedCuratorException {
		return initEnhancedCuratorFramework(partition, connectString, DEFAULT_SESSION_TIMEOUT_MS,
				DEFAULT_CONNECTION_TIMEOUT_MS, new RetryNTimes(DEFAULT_RETRY_TIMES, DEFAULT_SLEEPMS_BETWEENRETRY));
	}
	public static String getTargetConnectString(String path) throws EnhancedCuratorException {
		return matchBootstrapTargetConnectString(bootstrapTargetConnectString(path));
	}

	private static String bootstrapTargetConnectString(String path) throws EnhancedCuratorException {
		Map<String, Object> zkPropsMap = ZKUtils.propertiesToMap(getBootstrapData(false));
		for (Map.Entry<String, Object> entry : zkPropsMap.entrySet()) {
			if (ZKUtils.isMatch(entry.getKey(), path)) {
				return entry.getValue().toString();
			}
		}

		return ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_CONNECTION);
	}
	private static String matchBootstrapTargetConnectString(String targetConnectString) {
		return matchTargetConnectString(targetConnectString,
				ZKUtils.resolveSystemProperty(VIP_CFGCENTER_BOOTSTRAP_DEFAULT_MATCHER));
	}
	public static String getTargetConnectStringByDomain(String domain) throws EnhancedCuratorException {
		return getTargetConnectStringByDomain(getEnvPartition(), domain);
	}
	public static String getTargetConnectStringByDomain(String partition, String domain)
			throws EnhancedCuratorException {
		return matchDomainTargetConnectString(domainTargetConnectString(partition, domain));
	}
	private static String domainTargetConnectString(String partition, String domain) throws EnhancedCuratorException {
		return getConnStringByDomainData(getDomainData(partition, domain));
	}
	public static String getDomainData(String partition, String domain) throws EnhancedCuratorException {
		return getDomainData(ZKUtils.resolveSystemProperty(VIP_CFGCENTER_ZK_CONNECTION), partition, domain);
	}
	public static String getDomainData(String connectString, String partition, String domain)
			throws EnhancedCuratorException {
		EnhancedCuratorFramework tempEnhancedCuratorFramework = initTempEnhancedCuratorFramework(connectString,
				partition);
		try {
			tempEnhancedCuratorFramework.start();

			String clusterPath = null;
			final String domainZkPath = "/" + DOMAINS_MAPPING + "/" + domain;
			if (null != tempEnhancedCuratorFramework.checkExists().forPath(domainZkPath)) {
				byte[] bytes = tempEnhancedCuratorFramework.getData().forPath(domainZkPath);
				clusterPath = new String(bytes, DEFAULT_CHARSET);

				if (StringUtils.isBlank(clusterPath)) {
					throw new EnhancedCuratorException("zkdata in domainZkPath " + domainZkPath + " is null.");
				}
			}

			if (null == clusterPath) {
				final String defaultDomainPath = "/" + DOMAINS_MAPPING + "/" + DEFAULT_DOMAIN;
				if (null != tempEnhancedCuratorFramework.checkExists().forPath(defaultDomainPath)) {
					byte[] bytes = tempEnhancedCuratorFramework.getData().forPath(defaultDomainPath);
					clusterPath = new String(bytes, DEFAULT_CHARSET);

					if (StringUtils.isBlank(clusterPath)) {
						throw new EnhancedCuratorException("zkdata in defaultDomainPath " + defaultDomainPath
								+ " is null.");
					}
				}
			}

			if (StringUtils.isNotBlank(clusterPath)) {
				if (null != tempEnhancedCuratorFramework.checkExists().forPath(clusterPath)) {
					byte[] bytes = tempEnhancedCuratorFramework.getData().forPath(clusterPath);
					if (bytes != null && bytes.length > 0) {
						return new String(bytes, DEFAULT_CHARSET);
					} else {
						throw new EnhancedCuratorException("zkdata in clusterPath " + clusterPath + " is null.");
					}
				} else {
					throw new EnhancedCuratorException("clusterPath " + clusterPath + " is empty.");
				}
			} else {
				throw new EnhancedCuratorException("clusterPath is null.");
			}
		} catch (Exception e) {
			throw new EnhancedCuratorException(e);
		} finally {
			if (tempEnhancedCuratorFramework != null) {
				tempEnhancedCuratorFramework.close();
			}
		}
	}
	private static String matchDomainTargetConnectString(String targetConnectString) {
		return matchTargetConnectString(targetConnectString,
				ZKUtils.resolveSystemProperty(VIP_CFGCENTER_DOMAINS_DEFAULT_MATCHER));
	}
	
	public static String matchTargetConnectString(String targetConnectString, String defaultMatcherProperty) {
		String connectionMatcher = ZKUtils.resolveSystemProperty(VIP_CFGCENTER_CONNECTION_MATCHER);// 10.10.*.*
		if (StringUtils.isBlank(connectionMatcher)) {
			return useDefaultMatcher(targetConnectString, defaultMatcherProperty);
		}

		String target = matchPattern(targetConnectString, connectionMatcher);
		if (StringUtils.isNotBlank(target)) {
			return target;
		}

		return targetConnectString;
	}
	private static String useDefaultMatcher(String targetConnectString, String defaultMatcherProperty) {
		String target = "";
		boolean defaultMatcher = true;
		if (StringUtils.isNotEmpty(defaultMatcherProperty)) {
			defaultMatcher = Boolean.parseBoolean(defaultMatcherProperty);
		}

		if (defaultMatcher) {
			String ipPattern = "";
			try {
				String myIp = InetAddress.getLocalHost().getHostAddress();
				String[] ipArray = StringUtils.split(myIp, DOT);
				ipPattern = ipArray[0] + DOT + ipArray[1] + DOT + "**";
			} catch (Exception e) {
				LOGGER.error("useDefaultMatcher, get localhost error." + e.getMessage());
			}

			target = matchPattern(targetConnectString, ipPattern);
		}

		if (StringUtils.isNotBlank(target)) {
			return target;
		}

		return targetConnectString;
	}
	private static String matchPattern(String targetConnectString, String matchPattern) {
		StringBuilder targetBuilder = new StringBuilder();
		String[] zkConnectStrings = StringUtils.split(targetConnectString, COMMA);
		for (final String zkConnectString : zkConnectStrings) {
			if (ZKUtils.isMatch(matchPattern, zkConnectString)) {
				targetBuilder.append(zkConnectString + COMMA);
			}
		}

		return StringUtils.removeEnd(targetBuilder.toString(), COMMA);
	}
	public static String getTargetConnectStringByConnectDomain(String connectString, String domain)
			throws EnhancedCuratorException {
		return matchDomainTargetConnectString(domainTargetConnectString(connectString, getEnvPartition(), domain));
	}
	private static String domainTargetConnectString(String connectString, String partition, String domain)
			throws EnhancedCuratorException {
		return getConnStringByDomainData(getDomainData(connectString, partition, domain));
	}


}
